package juc;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;

/**
 * 生产者-消费者模型（Demo）
 * 生产者-生产10个对象供消费者消费，消费者消费完再生产10个对象，生产者生产完再消费，消费者消费完再退出。
 * TODO 待完善：进行代码标注，优化代码结构，增加日志输出，增加异常处理等。
 * 使用信号量的方式完成
 */

public class ProducerConsumerSemaphore {
    // 缓冲队列大小
    private static final int BUFFER_SIZE = 10;
    // 生产对象的最大数量
    private static final int MAX_ITEMS = 200;

    // 线程安全的队列
    private static final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(BUFFER_SIZE);
    // 信号量对象：初始许可数量为1的信号量
    private static final Semaphore producerSemaphore = new Semaphore(1);
    // 信号量对象：初始许可数量为0的信号量
    private static final Semaphore consumerSemaphore = new Semaphore(0);
    private static int producedCount = 0;
    private static int consumedCount = 0;

    public static void main(String[] args) {
        // 生产者
        Thread producerThread = new Thread(new Producer());
        // 消费者
        Thread consumerThread = new Thread(new Consumer());

        // 生产者工厂启动
        producerThread.start();
        // 消费者工厂启动
        consumerThread.start();
    }

    static class Producer implements Runnable {
        @Override
        public void run() {
            while (producedCount < MAX_ITEMS) {
                try {
                    producerSemaphore.acquire();
                    for (int i = 0; i < BUFFER_SIZE && producedCount < MAX_ITEMS; i++) {
                        int item = producedCount + 1;
                        queue.put(item);
                        System.out.println("Produced: " + item);
                        producedCount++;
                        Thread.sleep(2000);
                    }
                    consumerSemaphore.release();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            while (consumedCount < MAX_ITEMS) {
                try {
                    consumerSemaphore.acquire();
                    for (int i = 0; i < BUFFER_SIZE && consumedCount < MAX_ITEMS; i++) {
                        int item = queue.take();
                        System.out.println("Consumed: " + item);
                        consumedCount++;
                        Thread.sleep(2000);
                    }
                    producerSemaphore.release();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

